package org.bigdata.flink.example

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * @Author: chengjj
  * @Date: 2020-10-29
  */
object WordCount {
    def main(args: Array[String]): Unit = {
        // 定义一个数据类型保存单词出现的次数
        case class WordWithCount(word: String, count: Long)

        // 获取运行环境
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        // 连接此socket获取输入数据
        val text = env.socketTextStream("192.168.128.129", 9999)
        //需要加上这一行隐式转换 否则在调用flatmap方法的时候会报错
        // 解析数据, 分组, 窗口化, 并且聚合求SUM
        import org.apache.flink.api.scala._
        val windowCounts = text.flatMap(_.split(" "))
            //.map { w => WordWithCount(w, 1) }
            //.keyBy(_.word)
        //..timeWindow(Time.seconds(5), Time.seconds(1))
         //           .sum("count")
                .map { (_, 1) }
            .keyBy(_._1)
            .timeWindow(Time.seconds(5))
            .sum(1)
        // 打印输出并设置使用一个并行度
        windowCounts.print().setParallelism(1)
        env.execute("Socket Window WordCount")
    }
}
